<?php
namespace App\Command;


use Monolog\Logger;
use Monolog\Handler\StreamHandler;
use Doctrine\DBAL\DriverManager;
use Symfony\Component\Yaml\Yaml;
use Predis\Client as RedisClient;

/**
 * Class AbstractBase
 * @package App\Command
 */
abstract class AbstractBase

{
    /** @var  \Doctrine\DBAL\Connection ADS连接*/
    protected $conn;
    protected $queName;
    protected $msgQueue = array();
    protected $config;
    /** @var \AMQPQueue $channel */
    protected $channel;
    /** @var Logger $logger */
    protected $logger;
    /** @var \Doctrine\DBAL\Connection mysql连接 */
    protected $dbconn;
    /** @var RedisClient */
    protected   $redis;
    /** @var int 连接Rabbitmq失败次数 */
    private $tryConnectRabbitmqFailedTimes = 0;

    public function __construct($queName, $logInfo = 'cdr.consumer', $logFileName = 'cdr.log')
    {
        $arr = Yaml::parse(file_get_contents(__DIR__.'/../../config.yml'));
        $this->config = $arr['parameters'];
        $this->logger = new Logger($logInfo);
        $this->logger->pushHandler(new StreamHandler($this->config['log_dir'].'/'.$logFileName, Logger::INFO));
        $this->queName = $queName;
    }

    public function execute()
    {
        try {
            continue_connect:
//            if (!empty($this->conn)) {
//                $this->conn->close(); //关闭
//            }
            if (!empty($this->channel)) {
                $this->channel->getConnection()->disconnect(); //关闭
            }
            $this->channel = $this->getChannel(); //重连mq;
//            $this->conn = DriverManager::getConnection($this->config['ads']); //重连数据库
//            $this->conn->executeQuery("set names utf8");
            $this->dbconn = DriverManager::getConnection($this->config['mysql']); //连接数据库
            $this->dbconn->executeQuery("set names utf8");
            //链接redis
            if (!empty($this->redis)) {
                $this->redis->disconnect();
            }
            $this->redis = new RedisClient($this->config['redis']);//重连redis;
            continue_to_consume_data:
            $this->channel->consume(function (\AMQPEnvelope $envelope, \AMQPQueue $queue) {
                $body = $envelope->getBody();
                $tag = $envelope->getDeliveryTag();
                $this->logger->info(sprintf('[%s]处理消息【%s】', $this->queName, $body));
//                if ($this->ping() === false) {
//                    $this->conn->close();
//                    $this->conn->connect(); //重连；
//                }
                if ($this->dbconn->ping() === false) {
                    $this->dbconn->close();
                    $this->dbconn->connect(); //重连；
                }
                if (!empty($body)) {
                    $msgArray = json_decode($body, true);
                    if (json_last_error() != JSON_ERROR_NONE) {
                        $this->logger->error(
                            sprintf('队列【%s】解析消息【%s】失败，错误为【%s】,该消息不处理，直接ack', $this->queName, $body, json_last_error())
                        );
                        $queue->ack($tag);
                    } else {
                        $msgArray = $this->proccessMsgArray($msgArray);
                        // 放到待处理队列中，统一处理
                        $this->msgQueue[$tag] = $msgArray;
                    }
                } else {
                    $this->logger->error(sprintf('[%s]收到空消息，直接ack', $this->queName));
                    $queue->ack($tag);
                }

                /**
                 * 取出预设的数据条数再进行处理，如果数据不够的话，这种情况需要处理
                 */
                if (count($this->msgQueue) == $this->config['rabbitmq']['prefetch_count']) {
                    $this->consumerMessage($queue);
                }
            }, AMQP_NOPARAM);
        } catch (\AMQPQueueException $e) {
            //消费超时;read_timeout;
            $this->consumerMessage($this->channel); //超时就消费掉剩下的消息
            $this->logger->info(sprintf('que【%s】AMQPQueueException【%s】', $this->queName, $e->getMessage()));
            try {
                $this->channel->cancel();
            } catch (\Exception $e) {
                goto continue_connect;
            }
            goto continue_to_consume_data;
        } catch (\AMQPConnectionException $e) {
            $this->logger->info(sprintf('que【%s】AMQPConnectionException【%s】', $this->queName, $e->getMessage()));
            goto continue_connect;
        } catch (\AMQPException $e) {
            $this->logger->error(sprintf('que【%s】AMQPException【%s】', $this->queName, $e->getMessage()));
            goto continue_connect;
        } catch (\Exception $e) {
            $this->logger->error(sprintf(
                'que【%s】Exception【%s】class【%s】',
                $this->queName,
                $e->getMessage(),
                get_class($e)
            ));
            goto continue_connect;
        }
    }

    /**
     * 获取Rabbitmq连接
     *
     * @return \AMQPQueue $que
     */
    public function getChannel()
    {
        try {
            //创建连接和channel
            $connArgs = array(
                'host' => $this->config['rabbitmq']['host'],
                'port' => $this->config['rabbitmq']['port'],
                'login' => $this->config['rabbitmq']['user'],
                'password' => $this->config['rabbitmq']['password'],
                'vhost'=>'/',
                'read_timeout'=> $this->config['rabbitmq']['read_timeout'],
                'connect_timeout'=> $this->config['rabbitmq']['connect_timeout'],
                'heartbeat'=> $this->config['rabbitmq']['heartbeat'],
            );
            $connection = new \AMQPConnection($connArgs);
            if (!$connection->connect()) {
                $this->logger->error("不能够连接rabbitmq");
                $this->getChannel();
            }
            $channel = new \AMQPChannel($connection);
            $channel->qos(0, $this->config['rabbitmq']['prefetch_count']); //消费者处理最大条数数据；
            $que = new \AMQPQueue($channel);
            $que->setName($this->queName);

            $this->logger->info(sprintf(
                '[%s]第【%s】次连接Rabbitmq【%s:******@%s:%s】成功',
                $this->queName,
                $this->tryConnectRabbitmqFailedTimes,
                $this->config['rabbitmq']['user'],
                $this->config['rabbitmq']['host'],
                $this->config['rabbitmq']['port']
            ));
            $this->tryConnectRabbitmqFailedTimes = 0;
        } catch (\Exception $e) {
            $this->tryConnectRabbitmqFailedTimes++;
            $this->logger->error(sprintf(
                '[%s]第【%s】次连接Rabbitmq失败，错误为【%s】',
                $this->queName,
                $this->tryConnectRabbitmqFailedTimes,
                $e->getMessage()
            ));
            $sleepTime = $this->tryConnectRabbitmqFailedTimes*5;
            $sleepTime = $sleepTime < 300 ? $sleepTime : 300;
            sleep($sleepTime);

            return $this->getChannel();
        }

        return $que;
    }
    /**
     * ACk数据
     * @param \AMQPQueue $que
     * @param $key
     * @param $row
     * @param $excepNeedHandle
     */
    protected function ackData(\AMQPQueue $que, $key, $row, $excepNeedHandle = false)
    {
        //失败次数
        $sendFail = 0;
        //处理完毕之后回应ACK
        try {
            $que->ack($key);
            //删除掉ack的的
            $this->logger->info(sprintf(
                '[%s]ack数据【%s】成功',
                $this->queName,
                json_encode($row)
            ));
        } catch (\AMQPException $e) {
            //ack数据失败可能是
            $this->logger->error(sprintf(
                '[%s]ack数据【%s】失败,message【%s】',
                $this->queName,
                json_encode($row),
                $e->getMessage()
            ));
            //如果需要处理;发送到;
        }
    }

    /**
     * 消费方法
     * @param \AMQPQueue $que
     * @return mixed
     */
    abstract public function consumerMessage(\AMQPQueue $que);

    abstract public function proccessMsgArray($msgArray);
}
